# -*- coding: UTF-8 -*-
from DBUtils.PooledDB import PooledDB
import pymysql
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils

conn_pool = PooledDB(creator=pymysql,mincached = 5,maxcached=10,maxshared=50,blocking = True, maxconnections=50,host='h-master',user='root',passwd='123456',db='spark_kafka',port=3306)


def exec_sql(sql):
    conn = conn_pool.connection()
    cur = conn.cursor()
    try:
        print(sql)
        cur.execute(sql)
        conn.commit()
    except BaseException as e:
        print(e)
        conn.rollback()
    rs = cur.fetchone()
    cur.close()
    conn.close()
    if rs != None:
        return rs[0]
    else:
        return -1

def data_mysql(list):
    key = list[0].replace("\'","\\'")
    value = list[1]
    count = int(exec_sql("select countNum from key_value2 where keystr = '{}'".format(key)))
    if count > 0 and count != None:
        value = count+(int)(value)
        exec_sql("update  key_value2 set countNum = '{value}' where keystr = '{key}'".format(key=key,value=value))
    else:
        exec_sql("insert into key_value2 VALUES ('{key}','{value}')".format(key=key,value=value))
    pass


def data_to_mysql(rdd):
    taken = rdd.take(rdd.count())
    print("=========================开始=======================")
    for record in taken:
        print(str(record[0])+":"+str(record[1]))
        data_mysql(record)
    print("=========================结束=======================")


if __name__ == "__main__":

    # conn = pymysql.connect(host="localhost", user="root", passwd="123456", db="vinsinwsweb")
    topic_list =["kafkaSpark"]
    brokers = "h-master:9092"
    sc = SparkContext(appName="PythonStreamingDirectKafkaWordCount")
    ssc = StreamingContext(sc, 2)


    kvs = KafkaUtils.createDirectStream(ssc, topic_list, {"metadata.broker.list": brokers})

    lines = kvs.map(lambda x: x[1])
    counts = lines.flatMap(lambda line: line.split(" ")) \
        .map(lambda word: (word, 1)) \
        .reduceByKey(lambda a, b: a + b)
    counts.foreachRDD(data_to_mysql)
    #counts.pprint()
    ssc.start()
    ssc.awaitTermination()

    # print(exec_sql(conn_pool, "select password from users"))
